package org.pr.kafkaexpress.util;

import lombok.extern.slf4j.Slf4j;
import org.pr.kafkaexpress.constants.KafkaConstants;
import org.pr.kafkaexpress.exception.KafkaException;

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.UUID;

/**
 * 消息处理工具类
 * 
 * @author LXC
 * @since 2025/7/10
 */
@Slf4j
public final class MessageUtils {
    
    private static final DateTimeFormatter TIMESTAMP_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    
    private MessageUtils() {
        // 私有构造函数，防止实例化
    }
    
    /**
     * 验证消息是否有效
     * 
     * @param message 消息内容
     * @return 是否有效
     */
    public static boolean isValidMessage(String message) {
        if (message == null || message.trim().isEmpty()) {
            return false;
        }
        
        if (message.length() > KafkaConstants.MAX_MESSAGE_SIZE) {
            log.warn("Message size exceeds maximum allowed size: {} bytes", message.length());
            return false;
        }
        
        return true;
    }
    
    /**
     * 生成消息ID
     * 
     * @return 消息ID
     */
    public static String generateMessageId() {
        return UUID.randomUUID().toString();
    }
    
    /**
     * 生成带时间戳的消息
     * 
     * @param message 原始消息
     * @return 带时间戳的消息
     */
    public static String addTimestamp(String message) {
        String timestamp = LocalDateTime.now().format(TIMESTAMP_FORMATTER);
        return String.format("[%s] %s", timestamp, message);
    }
    
    /**
     * 计算消息的MD5哈希值
     * 
     * @param message 消息内容
     * @return MD5哈希值
     */
    public static String calculateMessageHash(String message) {
        try {
            MessageDigest md = MessageDigest.getInstance("MD5");
            byte[] hashBytes = md.digest(message.getBytes(StandardCharsets.UTF_8));
            StringBuilder sb = new StringBuilder();
            for (byte b : hashBytes) {
                sb.append(String.format("%02x", b));
            }
            return sb.toString();
        } catch (NoSuchAlgorithmException e) {
            log.error("MD5 algorithm not available", e);
            throw new KafkaException("Failed to calculate message hash", e);
        }
    }
    
    /**
     * 格式化消息用于日志输出
     * 
     * @param message 消息内容
     * @param maxLength 最大长度
     * @return 格式化后的消息
     */
    public static String formatMessageForLog(String message, int maxLength) {
        if (message == null) {
            return "null";
        }
        
        if (message.length() <= maxLength) {
            return message;
        }
        
        return message.substring(0, maxLength) + "...";
    }
    
    /**
     * 检查消息是否为心跳消息
     * 
     * @param message 消息内容
     * @return 是否为心跳消息
     */
    public static boolean isHeartbeatMessage(String message) {
        return message != null && message.trim().equalsIgnoreCase("heartbeat");
    }
    
    /**
     * 创建心跳消息
     * 
     * @return 心跳消息
     */
    public static String createHeartbeatMessage() {
        return addTimestamp("HEARTBEAT");
    }
} 